package com.clx.risk.flink.demo.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.clx.risk.flink.demo.conf.CommonConf;
import com.clx.risk.flink.demo.sinks.HashSetRedisMapper;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.redis.RedisSink;

public class ProcessWindowsCountWordApp {

    public static void main(String[] args) {
        try {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

            // source
            DataStream<String> stream = env.addSource(CommonConf.getKafkaConsumerSource());
            System.out.println(ProcessWindowsCountWordApp.class.getName() + "---开始接收数据");
            stream.print("kafka");

            //数据过滤
            DataStream<String> filterStream = filter(stream);

            // transform
            DataStream<Tuple2<String, Integer>> kvStream = filterStream.map(JSON::parseObject).map(new RichMapFunction<JSONObject, Tuple2<String, Integer>>() {

                @Override
                public Tuple2 map(JSONObject jsonObject) throws Exception {
                    String productcode = jsonObject.getString("productCode");
                    return new Tuple2<String, Integer>(productcode, 1);
                }
            });


            KeyedStream<Tuple2<String, Integer>, Tuple> groupStream = kvStream.keyBy(0);
            SingleOutputStreamOperator<Tuple2<String, Integer>> sumStream = groupStream.timeWindow(Time.days(1), Time.seconds(5)).sum(1);

            // 将结果写入redis
            sumStream.addSink(new RedisSink(CommonConf.getFlinkJedisPoolConfig(), new HashSetRedisMapper()));

            // 执行任务
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static DataStream<String> filter(DataStream<String> source) {
        return source.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String productCode = jsonObject.getString("productCode");
                if (productCode == null || productCode.trim().length() <= 1) {
                    return false;
                } else {
                    return true;
                }
            }
        });
    }
}
